<?php
namespace app\process;

use Swoole\Coroutine;
use Swoole\Coroutine\Channel;
use Simps\MQTT\Client;
use Simps\MQTT\Config\ClientConfig;
use app\process\MqttDataHandle;

class MqttLinuxListener
{
    // MQTT 服务器配置
    private $mqttConfig = [
        'host' => 'dev.fepft.com', // MQTT 服务器地址
        'port' => 1883, // 端口（加密连接用8883）
        'client_id' => 'webman_mqtt_', // 客户端ID（需唯一）
        'username' => 'dingweiqi', // 认证用户名（可选）
        'password' => 'maxhom168', // 认证密码（可选）
        'keepalive' => 100, // 心跳间隔（秒）
    ];

    // 要监听的主题（支持通配符），主题 => QoS等级
    private $topics = [
        '/server/#' => 0,
        // 下面是系统主题，订阅客户端断开连接的消息
        // '$SYS/brokers/+/clients/+/connected' => 0,
        // '$SYS/brokers/+/clients/+/disconnected' =>0,
    ];

    private MqttDataHandle $handler;              // 业务处理类
    private Channel $taskChan;                    // 任务队列
    private Client $mqttClient;                   // MQtt客户端
    private int $coNum = 5;                       // 只开 5 条消费协程

    public function onWorkerStart($worker): void
    {
        $this->handler = new MqttDataHandle();
        // 必须在调度器里
        Coroutine\run(fn() => $this->run());
    }


    private function run(): void
    {
        // 任务队列（容量略大于协程数即可）
        $this->taskChan = new Channel($this->coNum + 20);

        // 启动「纯消费者」协程（只读队列）
        for ($i = 0; $i < $this->coNum; $i++) {
            go(fn() => $this->consumer());
        }

        // 主协程：接受MQTT数据并写入队列
        $this->receiver();
    }

    // 生产者，接收MQTT数据并写入队列
    private function receiver(): void{
        $config = (new ClientConfig())
            ->setUserName($this->mqttConfig['username'])
            ->setPassword($this->mqttConfig['password'])
            ->setClientId($this->mqttConfig['client_id'] . uniqid())
            ->setKeepAlive($this->mqttConfig['keepalive'])
            ->setDelay(5000) // 连接断开后，等待多久开始第一次重连, 这里设置为5s
            ->setMaxAttempts(0) // 设置为 0 表示连接断开后无限次重试，直到重连成功（生产环境常用配置）
            ->setSwooleConfig([
                'open_mqtt_protocol' => true, // 必须设为 true，否则 Swoole 无法解析 MQTT 协议格式的数据包
                'package_max_length' => 2 * 1024 * 1024, // 最大包长度（2MB）
                'connect_timeout' => 5.0,   // 连接超时时间（5秒）
                'write_timeout' => 5.0,     // 写入数据超时时间（5秒）
                'read_timeout' => 5.0,      // 读取数据超时时间（5秒）
            ]);


        $mqtt = new Client('dev.fepft.com', 1883, $config);
        // 以清除会话模式连接 MQTT 服务器，并设置遗嘱消息 —— 若客户端异常断开，服务器会自动发布 byebye 到 设置的topic主题
        $mqtt->connect(true, [
            'topic' => '/server_delete',
            'qos' => 0,
            'retain' => 0,
            'message' => 'byebye',
        ]);
        // 订阅配置
        $mqtt->subscribe($this->topics);
        $this->mqttClient = $mqtt;

        while (true) {
            $data = $mqtt->recv();
            if ($data) {
                // 非阻塞 push，如果队列满会挂起 50ms 再试
                if ($data && $data !== true) {
                    while (!$this->taskChan->push($data)) {
                        Coroutine::sleep(0.05);
                    }
                }
            }
        }
    }

    // 消费者循环获取消息
    private function consumer(): void
    {
        $cid = Coroutine::getCid();
        while (true) {
            $task = $this->taskChan->pop();   // 没有任务就挂起
	        $this->handle($task['topic'], $task['message'], $task, $cid);
        }
    }

    // 消费者逻辑
    private function handle($topic, $payload, $conetnt, $cid): void
    {
        printMqttLog("协程- <$cid> 开始处理 $topic 消息: ". json_encode($conetnt));
        $this->handler->handleData($topic, $payload, $this->mqttClient);
    }
}


